package com.pangyangyang.elasticsearch.plugin.canal.collector;

import com.alibaba.otter.canal.client.impl.SimpleCanalConnector;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.alibaba.otter.canal.protocol.Message;
import com.alibaba.otter.canal.protocol.exception.CanalClientException;
import com.google.protobuf.InvalidProtocolBufferException;
import com.pangyangyang.elasticsearch.plugin.util.MessageRiverConstants;
import com.pangyangyang.elasticsearch.plugin.dispatch.EventDispatcherHolder;
import com.pangyangyang.elasticsearch.plugin.event.DataModifyEvent;
import com.pangyangyang.elasticsearch.plugin.event.DataModifyEventType;
import com.pangyangyang.elasticsearch.plugin.river.Build;
import com.pangyangyang.elasticsearch.plugin.river.MessageRiverSettings;
import org.elasticsearch.common.logging.ESLogger;
import org.elasticsearch.common.logging.ESLoggerFactory;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;

/**
 * 当前只支持单线程执行，问题在于如何区分Transaction的开始和结束位置
 * Created by yangwang on 2015/1/15.
 */
public class CanalEventTask implements Runnable {

    private SimpleCanalConnector connector;
    private boolean blocked;
    private int batchSize;
    private int sleepSeconds;
    private long timeoutSeconds;
    private CanalCollectorFailOverStrategy failOverStrategy;

    private boolean transactionBegin = false;
    private  boolean stop = false;
    private List<DataModifyEvent> cachedEvents = new ArrayList<DataModifyEvent>();
    private List<DataModifyEvent> dispatchEvents = new ArrayList<DataModifyEvent>();

    private static ESLogger logger = ESLoggerFactory.getLogger(Build.LOGGER_NAME);

    public CanalEventTask(SimpleCanalConnector connector,  MessageRiverSettings settings) {
        this.connector = connector;
        blocked = settings.getBoolean(MessageRiverConstants.SETTING_KEY_CANAL_BLOCKED, false);
        batchSize = settings.getInt(MessageRiverConstants.SETTING_KEY_CANAL_BATCH_SIZE, 1024);
        sleepSeconds = settings.getInt(MessageRiverConstants.SETTING_KEY_CANAL_SLEEP_SEC, 1);
        timeoutSeconds = settings.getInt(MessageRiverConstants.SETTING_KEY_COMMON_TIMEOUT, 30);
        this.failOverStrategy = new DefaultFailOverStrategy(this);
    }

    @Override
    public void run() {
        try{
            failOverStrategy.start(connector);
            while(!failOverStrategy.shouldStop()){
                Message message = null;
                if (blocked){
                    message = connector.getWithoutAck(batchSize, timeoutSeconds, TimeUnit.SECONDS);
                } else {
                    message = connector.getWithoutAck(batchSize);
                }

                handleMessage(message);
                dispatchEvent();
                if(message.getId() != -1)
                    logger.debug("ack message {}", message.getId());

                connector.ack(message.getId());

                if(!blocked && sleepSeconds > 0){
                    try{
                        Thread.sleep(sleepSeconds * 1000);
                    }catch (InterruptedException e){
                        logger.warn("canal collector sleep interrupted, thread continue run.", e);
                    }
                }
            }
        } catch (Throwable t){
            failOverStrategy.whenError(connector, t);
        } finally {
            failOverStrategy.stop(connector);
        }
    }

    public void stop(){
        stop = true;
    }

    public boolean isStop() {
        return stop;
    }

    protected void dispatchEvent(){
        List<Future> futures = new ArrayList<Future>();
        if(dispatchEvents.isEmpty()){
            return;
        }

        logger.debug("dispatching canal event size {}.", dispatchEvents.size());
        for (DataModifyEvent event : dispatchEvents){
            Future future = EventDispatcherHolder.dispatchEvent(event);
            futures.add(future);
        }

        //wait finished
        for (Future future : futures){
            try {
                future.get();
            } catch (Exception e) {
                logger.error("Dispatch event wait finish failed.", e);
            }
        }
        dispatchEvents.clear();
        logger.debug("dispatching is finished.");
    }

    protected void handleMessage(Message message){
        int size = (message.getEntries() == null) ? 0 : message.getEntries().size();
        if(message.getId() == -1 || size == 0){
            //do nothing
        } else {
            for (CanalEntry.Entry entry : message.getEntries()){
                switch (entry.getEntryType()){
                    case TRANSACTIONBEGIN:
                        transactionBegin = true;
                        break;
                    case TRANSACTIONEND:
                        transactionBegin = false;
                        //以commit时间为准把数据加入到dispatchList中
                        break;
                    case HEARTBEAT:
                        break;
                    case ROWDATA:
                        handleRowData(entry);
                        break;
                }
                //TODO transaction rollback
            }
        }
        dispatchEvents.addAll(cachedEvents);
        cachedEvents.clear();
    }

    protected void handleRowData(CanalEntry.Entry entry){
        String destination = getDestination(entry);
        DataModifyEvent event = null;
        try {
            CanalEntry.RowChange rowChange = CanalEntry.RowChange.parseFrom(entry.getStoreValue());
            CanalEntry.EventType eventType = rowChange.getEventType();

            for (CanalEntry.RowData rowData : rowChange.getRowDatasList()) {
                if (eventType == CanalEntry.EventType.DELETE) {
                    event = scanColumn(destination, eventType, rowData.getBeforeColumnsList());
                } else if (eventType == CanalEntry.EventType.INSERT) {
                    event = scanColumn(destination, eventType, rowData.getAfterColumnsList());
                } else {
                    event = scanColumn(destination, eventType, rowData.getAfterColumnsList());
                }
                cachedEvents.add(event);
            }
        } catch (InvalidProtocolBufferException e) {
            logger.warn("Failed to parse entry {} - {} - {}.", e, entry.getHeader().getSchemaName(), entry.getHeader().getTableName(), entry.getEntryType());
        }
    }

    protected  DataModifyEvent scanColumn(String destination, CanalEntry.EventType type, List<CanalEntry.Column> columnList){
        DataModifyEvent event = new DataModifyEvent();
        event.setDataDestination(destination);

        switch (type){
            case INSERT:
                event.setType(DataModifyEventType.INSERT);
                break;
            case UPDATE:
                event.setType(DataModifyEventType.UPDATE);
                break;
            case DELETE:
                event.setType(DataModifyEventType.DELETE);
                break;
        }

        for (CanalEntry.Column column : columnList){
            event.putData(column.getName(), column.getValue());
        }

        return event;
    }

    private String getDestination(CanalEntry.Entry entry){
        String schema = entry.getHeader().getSchemaName();
        String table = entry.getHeader().getTableName();

        return "canal" + MessageRiverConstants.COMMON_DOT +  schema + MessageRiverConstants.COMMON_DOT + table;
    }
}
